#ifndef _NETWORK_THREAD_HPP
#define _NETWORK_THREAD_HPP
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/steady_timer.hpp>
#include <cstdint>
#include <memory>
#include <thread>
#include <mutex>
#include <list>
#include <atomic>
#include <iostream>
#include "Socket.hpp"
using boost::asio::ip::tcp;

template <class SocketType>
class NetworkThread
{
public:
    NetworkThread() : _acceptSocket(_io_context), _thread(nullptr), _updateTimer(nullptr), _connected(0), _stoped(false){}
    virtual ~NetworkThread()
    {
        if(_updateTimer)
        {
            delete _updateTimer;
            _updateTimer = nullptr;
        }

        if(_thread)
        {
            Stop();
            _thread->join();
            delete _thread;
            _thread = nullptr;
        }
    }
public:
    //  virtual function
    virtual void AddSocketed(std::shared_ptr<SocketType> new_sc) {}

    virtual void RemoveSocketed(std::shared_ptr<SocketType> rm_sc) {}
public:
    // real socket type
    typedef SocketType _SocketType;

    void Start()
    {
        assert(!_updateTimer && !_thread);

        _updateTimer = new boost::asio::steady_timer(_io_context, boost::asio::chrono::milliseconds(1));

        _thread = new std::thread([this](){
            this->Run();
        });
    }

    void AddSocket(tcp::socket&& new_sc)
    {
        std::lock_guard<std::mutex> guard(_new_socket_mutex);
        std::shared_ptr<SocketType> sc = std::make_shared<SocketType>(std::forward<tcp::socket>(new_sc));
        if(sc->IsOpen())
        {
            _newSockets.push_back(sc);
        }
    }

    void Stop()
    {
        if(_stoped.exchange(true)) return;

        _io_context.stop();
    }

    uint32_t GetNetWorkLoad()
    {
        return _connected;        
    }

    tcp::socket& GetSocket()
    {
        return _acceptSocket;
    }

    void Update()
    {
        // 停止了就不处理了，正常逻辑一般是从业务层先stop，再到网络层
        if(_stoped) return;

        // pop new_socket
        {
            std::lock_guard<std::mutex> guard(_new_socket_mutex);
            for(std::shared_ptr<SocketType> it : _newSockets)
            {
                if(it->IsOpen())
                {
                    _sockets.push_back(it);
                    it->Start();
                    AddSocketed(it);
                    ++_connected;                
                }
            }
            _newSockets.clear();
        }

        _updateTimer->expires_after(boost::asio::chrono::milliseconds(1));
        _updateTimer->async_wait([this](boost::system::error_code const&){
            this->Update();
        });

        _sockets.erase(std::remove_if(_sockets.begin(), _sockets.end(), [this](std::shared_ptr<SocketType> sock)
        {
            if (!sock->Update())
            {
                if (sock->IsOpen())
                    sock->CloseSocket();

                this->RemoveSocketed(sock);

                --this->_connected;
                return true;
            }

            return false;
        }), _sockets.end());
    }
private:
    void Run()
    {
        _updateTimer->async_wait([this](boost::system::error_code const&){
            this->Update();
        });

        _io_context.run();

        // stop
        _newSockets.clear();
        _sockets.clear();
    }
private:
    // _io_context
    boost::asio::io_context _io_context;
    // _acceptSocket
    tcp::socket _acceptSocket;    
    // _thread
    std::thread* _thread;
    // update timer
    boost::asio::steady_timer* _updateTimer;
    // _mutex
    std::mutex _new_socket_mutex;
    // list<Socket> _newSockets
    std::list<std::shared_ptr<SocketType>> _newSockets;
    // List<Socket> _sockets
    std::list<std::shared_ptr<SocketType>> _sockets;
    // atomic<bool> _stoped
    std::atomic<bool> _stoped;
    // load
    std::atomic<uint16_t> _connected;
};

#endif